#include "MsgQueue.h"
#include <vector>
#include <iostream>
#include <unistd.h>
#include "util.h"
#include "Job.h"

using namespace std;

namespace PFS
{
    MsgQueue::MsgQueue(struct event_base *base)//, callback func)
    {
        int fds[2];
        if(pipe(fds) == -1)
        {
            fprintf(stderr,"pipe error.\n");
            this = NULL;
        }
        else
        {
            receive_fd = fds[0];
            send_fd = fds[1];
            event_set(&ev, receive_fd, EV_READ | EV_PERSIST, MsgQueue::pop, this);
            event_base_set(base, &ev);
            event_add(&ev, NULL);
        }
    }

    MsgQueue::~MsgQueue()
    {
        for(;queue.size > 0;)
            sleep(1);
        event_del(&ev);
        close(push_fd);
        close(pop_fd);
    }

    int MsgQueue::push(void* msg)
    {
        const char buf[1] = {0};
        mutex.lock();
        queue.push(msg);
        mutex.unlock();
        if(write(send_fd, buf, 1) != 1)
        {
            ceer<<"write pipe fd error!\n";
            return -1;
        }
        return 0;
    }

    int MsgQueue::length()
    {
        mutex.lock()
        int len = static_cast<int>(queue.size()); 
        mutex.unlock();
        return len;
    }

    void MsgQueue::pop(int fd, short flags, void *arg)
    {
        MsgQueue *msg = (MsgQueue *)arg;
        bool deal = true;
        char buf[1];
        return_if_fail(fd != msg->receive_fd);
        int ret = read(msg->receive_fd, buf, 1);
        return_if_fail(ret != 1);

        msg->mutex.lock();
        do
        {
            if(msg->queue.size() == 0)
            {
                deal = false;
                break;
            }
            struct event_arg* data = (struct event_arg*)msg->queue.front();
            msg->queue.pop();
            break;
        }while(0);
        msg->mutex.unlock();
        return_if_fail(deal == true);
        Job* job = data->job;
        job->run((void*) data);//类似于conn_new,通过data来获取一个conn的结构，并增加相应的event事件，然后进去相应的event事件，在那里面根据状态来做相应的动作。
    }
}
